package com.liqw;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RScript;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 分布式限流器 - 基于Redis Sorted Set实现
 * 支持并发控制、持有者记录、自动续期和过期清理
 */
@Slf4j
@Component
public class DistributedRateLimiter {
    // ==================== 常量定义 ====================
    private static final String PREFIX = "rate_limiter:";
    public static final String MAX_CONCURRENT_KEY_TEMPLATE = PREFIX + "max_concurrent:%s";
    public static final String HOLDERS_KEY_TEMPLATE = PREFIX + "holders:%s";

    // 默认过期时间配置
    public static final long MAX_CONCURRENT_KEY_DEFAULT_EXPIRE_DAYS = 1; // 最大并发数配置默认1天过期
    public static final long HOLDER_DEFAULT_EXPIRE_HOURS = 1; // 持有者默认1小时过期

    // 新增：退避重试的默认延迟时间（毫秒）
    public static final long DEFAULT_RETRY_DELAY_MILLIS = 50;
    // 定时清理任务间隔（毫秒）
    public static final long CLEANUP_INTERVAL_MILLIS = 5 * 60 * 1000; // 5分钟

    // 续期任务间隔：30 分钟（小于 HOLDER_DEFAULT_EXPIRE_HOURS 的 1 小时，确保在过期前续期）
    public static final long RENEW_LEASE_INTERVAL_MILLIS = 30 * 60 * 1000;
    // 示例定义：30 分钟
    static final long RENEW_INTERVAL_SECONDS = 30 * 60;


    /**
     * 合并后的本地缓存。
     * Key: holders_key (e.g., "rate_limiter:holders:my_service")
     * Value: 一个 ConcurrentHashMap，存储需要续期的 holder_id 及其最后续期时间戳。
     *        - Key: holder_id (e.g., "worker_thread_1")
     *        - Value: last_renew_timestamp (long)
     * 缓存项在写入后1小时未被访问则自动过期，避免内存泄漏。
     */
    private final LoadingCache<String, ConcurrentHashMap<String, Long>> activeHoldersCache = CacheBuilder.newBuilder()
            .expireAfterWrite(1, TimeUnit.HOURS)
            .maximumSize(10000)
            .build(new CacheLoader<String, ConcurrentHashMap<String, Long>>() {
                @Override
                public ConcurrentHashMap<String, Long> load(String holdersKey) {
                    // 当缓存中没有该 key 时，自动加载一个新的空 ConcurrentHashMap
                    return new ConcurrentHashMap<>();
                }
            });


    // ==================== 成员变量 ====================
    @Autowired
    private RedissonClient redissonClient;
    private String updateMaxConcurrentScript;
    private String acquireScript;
    private String releaseScript;
    private String getCurrentCountScript;
    private String getCurrentHoldersScript;
    private String cleanupScript;
    private String batchRenewLeaseScript;

    /**
     * 在 Bean 初始化后执行，用于构建并预编译 Lua 脚本
     */
    @PostConstruct
    public void initScripts() {
        this.updateMaxConcurrentScript = buildUpdateMaxConcurrentScript();
        this.acquireScript = buildAcquireScript();
        this.releaseScript = buildReleaseScript();
        this.getCurrentCountScript = buildGetCurrentCountScript();
        this.getCurrentHoldersScript = buildGetCurrentHoldersScript();
        this.cleanupScript = buildCleanupScript();
        this.batchRenewLeaseScript = buildBatchRenewLeaseScript();
    }

    private String buildUpdateMaxConcurrentScript() {
        StringJoiner joiner = new StringJoiner("\n");
        joiner.add("local max_key = KEYS[1]");
        joiner.add("local holders_key = KEYS[2]");
        joiner.add("local new_max = tonumber(ARGV[1])");
        joiner.add("local expire_seconds = tonumber(ARGV[2])");
        joiner.add("redis.call('set', max_key, new_max)");
        joiner.add("redis.call('expire', max_key, expire_seconds)");
        joiner.add("redis.call('expire', holders_key, expire_seconds)");
        joiner.add("return 1");
        return joiner.toString();
    }

    /**
     * 使用 StringJoiner 构建获取许可的 Lua 脚本
     * 新增：holders_key与max_key设置相同的过期时间并同步续期
     */
    private String buildAcquireScript() {
        StringJoiner joiner = new StringJoiner("\n");
        joiner.add("local holders_key = KEYS[1]");
        joiner.add("local max_key = KEYS[2]");
        joiner.add("local max_key_expire = tonumber(ARGV[1])");
        joiner.add("local expire_seconds = tonumber(ARGV[2])"); // 过期秒数，不再是时间戳
        joiner.add("local holder_id = ARGV[3]");

        // 直接在Lua中获取当前时间
        joiner.add("local current_seconds = tonumber(redis.call('time')[1])");
        joiner.add("local max_key_expire_timestamp = current_seconds + max_key_expire");
        joiner.add("local expire_timestamp = current_seconds + expire_seconds");

        // 1. 获取最大并发数，若不存在或无法转换为数字，返回 -1（获取失败）
        joiner.add("local max = redis.call('get', max_key)");
        joiner.add("if not max then");
        joiner.add("  return -1");
        joiner.add("end");
        joiner.add("max = tonumber(max)");
        joiner.add("if not max then return -1 end");  // max 无法转换为数字

        // 为最大并发数配置键设置过期时间（自动续期）
        joiner.add("redis.call('expire', max_key, max_key_expire_timestamp)");
        // 同时为持有者集合键设置相同的过期时间（同步续期）
        joiner.add("redis.call('expire', holders_key, max_key_expire_timestamp)");

        // 2. 获取当前有效持有者数量（不包括过期的）
        joiner.add("local current = redis.call('zcount', holders_key, current_seconds, '+inf')");

        // 3. 并发数比较
        joiner.add("if current >= max then");
        joiner.add("    return 0");
        joiner.add("end");

        // 4. 添加/更新持有者信息并返回成功标识
        joiner.add("redis.call('zadd', holders_key, expire_timestamp, holder_id)");
        joiner.add("return 1");
        return joiner.toString();
    }

    /**
     * 使用 StringJoiner 构建释放许可的 Lua 脚本
     */
    private String buildReleaseScript() {
        StringJoiner joiner = new StringJoiner("\n");
        joiner.add("local holders_key = KEYS[1]");
        joiner.add("local holder_id = ARGV[1]");

        // 移除指定持有者
        joiner.add("local removed = redis.call('zrem', holders_key, holder_id)");

        // 返回是否成功移除
        joiner.add("return removed");
        return joiner.toString();
    }

    /**
     * 构建获取当前并发数的 Lua 脚本
     * 只计算有效（未过期）的持有者
     */
    private String buildGetCurrentCountScript() {
        StringJoiner joiner = new StringJoiner("\n");
        joiner.add("local holders_key = KEYS[1]");
        joiner.add("local current_seconds = tonumber(redis.call('time')[1])");
        joiner.add("return redis.call('zcount', holders_key, current_seconds, '+inf')");
        return joiner.toString();
    }

    /**
     * 构建获取有效持有者列表的 Lua 脚本
     */
    private String buildGetCurrentHoldersScript() {
        StringJoiner joiner = new StringJoiner("\n");
        joiner.add("local holders_key = KEYS[1]");
        joiner.add("local current_seconds = tonumber(redis.call('time')[1])");
        joiner.add("return redis.call('zrangebyscore', holders_key, current_seconds, '+inf')");
        return joiner.toString();
    }

    /**
     * 构建清理过期持有者的 Lua 脚本
     * 直接在Lua中获取当前时间
     */
    private String buildCleanupScript() {
        StringJoiner joiner = new StringJoiner("\n");
        joiner.add("local holders_key = KEYS[1]");
        joiner.add("local current_seconds = tonumber(redis.call('time')[1])");

        // 删除所有过期的元素（score小于当前时间）
        joiner.add("local removed_count = redis.call('zremrangebyscore', holders_key, '-inf', current_seconds)");

        // 删除空集合
        joiner.add("if redis.call('zcard', holders_key) == 0 then");
        joiner.add("    redis.call('del', holders_key)");
        joiner.add("end");

        joiner.add("return removed_count");
        return joiner.toString();
    }

    /**
     * 构建批量 Holder 续期脚本。
     * 该脚本接收一个 holder_id 列表，并为其中每个存在且未过期的 Holder 进行续期。
     *
     * @return Lua 脚本字符串
     */
    private String buildBatchRenewLeaseScript() {
        StringJoiner joiner = new StringJoiner("\n");

        // 1. 定义输入参数
        // KEYS[1] -> holders_key (e.g., "rate_limiter:holders:my_service")
        // ARGV[1] -> expire_seconds (续期时长，单位：秒)
        // ARGV[2...] -> holder_ids (需要续期的 holder_id 列表)
        joiner.add("local holders_key = KEYS[1]");
        joiner.add("local expire_seconds = tonumber(ARGV[1])");
        joiner.add("local max_key_expire = tonumber(ARGV[2])");

        // 2. 获取当前时间戳（秒）
        joiner.add("local current_seconds = tonumber(redis.call('time')[1])");

        // 3. 初始化成功续期的计数器
        joiner.add("local renewed_count = 0");

        // 4. 遍历 holder_id 列表进行续期
        // ipairs 从索引 1 开始，但我们的数据从 ARGV[2] (索引 2) 开始
        joiner.add("for i = 2, #ARGV do");
        joiner.add("    local holder_id = ARGV[i]");

        // 4.1. 检查 Holder 是否存在且未过期
        joiner.add("    local existing_expire_time = redis.call('zscore', holders_key, holder_id)");
        joiner.add("    if existing_expire_time and tonumber(existing_expire_time) > current_seconds then");

        // 4.2. 计算并更新新的过期时间戳
        joiner.add("        local new_expire_time = current_seconds + expire_seconds");
        joiner.add("        redis.call('zadd', holders_key, new_expire_time, holder_id)");

        // 4.3. 增加成功计数器
        joiner.add("        renewed_count = renewed_count + 1");
        joiner.add("    end");
        joiner.add("end");

        // 5. 为整个 holders_key 续期 (如果有任何一个续期成功)
        // 这是一个优化，可以避免在没有任何续期操作时也调用一次 expire
        joiner.add("local max_key_expire = tonumber(ARGV[1])");
        joiner.add("local max_key_expire_timestamp = current_seconds + max_key_expire");
        joiner.add("if renewed_count > 0 then");
        joiner.add("    redis.call('expire', holders_key, max_key_expire_timestamp)");
        joiner.add("end");

        // 6. 返回成功续期的 Holder 数量
        joiner.add("return renewed_count");
        return joiner.toString();
    }

    // ==================== 核心方法 ====================

    /**
     * 尝试获取并发许可（非阻塞）
     */
    public boolean tryAcquire(String limiterKey, String holderId) {
        return tryAcquire(limiterKey, holderId, true);
    }

    /**
     * 尝试获取并发许可（非阻塞）
     */
    public boolean tryAcquire(String limiterKey, String holderId, boolean needRenew) {
        Assert.hasText(limiterKey, "限流器键(limiterKey)不能为空");
        Assert.hasText(holderId, "持有者ID(holderId)不能为空");

        String holdersKey = String.format(HOLDERS_KEY_TEMPLATE, limiterKey);
        String maxConcurrentKey = String.format(MAX_CONCURRENT_KEY_TEMPLATE, limiterKey);

        // 关键：获取或创建 holdersKey 对应的 ConcurrentHashMap
        // 这一步会自动将 holdersKey 加入活跃缓存
        ConcurrentHashMap<String, Long> renewingHolders = activeHoldersCache.getUnchecked(holdersKey);

        List<Object> keys = new ArrayList<>(2);
        keys.add(holdersKey);
        keys.add(maxConcurrentKey);

        // 转换过期时间为秒
        Long maxKeyExpireSeconds = TimeUnit.DAYS.toSeconds(MAX_CONCURRENT_KEY_DEFAULT_EXPIRE_DAYS);
        // 传递过期秒数，而不是时间戳
        Long expireSeconds = TimeUnit.HOURS.toSeconds(HOLDER_DEFAULT_EXPIRE_HOURS);

        RScript script = redissonClient.getScript();
        Long result = script.eval(
                RScript.Mode.READ_WRITE,
                acquireScript,
                RScript.ReturnType.INTEGER,
                keys,
                maxKeyExpireSeconds.intValue(), expireSeconds.intValue(), holderId
        );

        boolean success = result != null && result == 1;
        if (success && needRenew) {
            // 如果需要续期，则将 holderId 加入续期列表，并记录当前时间戳作为初始续期时间
            renewingHolders.put(holderId, System.currentTimeMillis());
        }

        // 如果获取失败且之前是续期holder，则移除
        if (!success && renewingHolders.containsKey(holderId)) {
            renewingHolders.remove(holderId);
        }

        return result != null && result == 1;
    }

    /**
     * 尝试获取并发许可（阻塞带超时）
     * 不清理过期持有者
     */
    public boolean tryAcquire(String limiterKey, String holderId, long timeout, TimeUnit unit) throws InterruptedException {
        Assert.hasText(limiterKey, "限流器键(limiterKey)不能为空");
        Assert.hasText(holderId, "持有者ID(holderId)不能为空");
        Assert.isTrue(timeout >= 0, "超时时间不能为负数");
        Assert.notNull(unit, "时间单位不能为null");

        // 快速尝试一次
        if (tryAcquire(limiterKey, holderId, false)) {
            return true;
        }

        if (timeout == 0) {
            return false;
        }

        long startTime = System.currentTimeMillis();
        long timeoutMillis = unit.toMillis(timeout);
        long retryDelay = DEFAULT_RETRY_DELAY_MILLIS;

        long elapsedTime = System.currentTimeMillis() - startTime;
        while (elapsedTime < timeoutMillis) {
            // 计算下一次应该等待的时间
            long timeLeft = timeoutMillis - elapsedTime;
            long sleepTime = Math.min(retryDelay, timeLeft);

            Thread.sleep(sleepTime);

            if (tryAcquire(limiterKey, holderId, false)) {
                return true;
            }

            // 实现指数退避，减少高并发下的Redis压力
            retryDelay = Math.min(retryDelay * 2, 1000); // 最多等待1秒

            elapsedTime = System.currentTimeMillis() - startTime;
        }

        return false;
    }

    /**
     * 释放并发许可
     * 会清理过期持有者
     */
    public boolean release(String limiterKey, String holderId) {
        Assert.hasText(limiterKey, "限流器键(limiterKey)不能为空");
        Assert.hasText(holderId, "持有者ID(holderId)不能为空");

        String holdersKey = String.format(HOLDERS_KEY_TEMPLATE, limiterKey);

        // 从续期列表中移除 holderId
        try {
            ConcurrentHashMap<String, Long> renewingHolders = activeHoldersCache.get(holdersKey);
            if (renewingHolders != null) {
                renewingHolders.remove(holderId);
            }
        } catch (Exception e) {
            log.warn("Failed to remove holder '{}' from renewing list for key '{}'. It might have expired already.", holderId, holdersKey, e);
        }

        List<Object> keys = new ArrayList<>(1);
        keys.add(holdersKey);

        RScript script = redissonClient.getScript();
        Long result = script.eval(
                RScript.Mode.READ_WRITE,
                releaseScript,
                RScript.ReturnType.INTEGER,
                keys,
                holderId
        );

        return result != null && result > 0;
    }

    /**
     * 获取当前有效并发数
     * 只计算未过期的持有者
     */
    public int getCurrentCount(String limiterKey) {
        Assert.hasText(limiterKey, "限流器键(limiterKey)不能为空");

        String holdersKey = String.format(HOLDERS_KEY_TEMPLATE, limiterKey);

        RScript script = redissonClient.getScript();
        Long result = script.eval(
                        RScript.Mode.READ_ONLY,
                        getCurrentCountScript,
                        RScript.ReturnType.INTEGER,
                        Collections.singletonList(holdersKey)
                );

        return result != null ? result.intValue() : 0;
    }

    /**
     * 获取当前有效持有者列表
     * 只返回未过期的持有者
     */
    public List<String> getCurrentHolders(String limiterKey) {
        Assert.hasText(limiterKey, "限流器键(limiterKey)不能为空");

        String holdersKey = String.format(HOLDERS_KEY_TEMPLATE, limiterKey);

        RScript script = redissonClient.getScript();
        List<String> result = script.eval(
                            RScript.Mode.READ_ONLY,
                            getCurrentHoldersScript,
                            RScript.ReturnType.INTEGER,
                            Collections.singletonList(holdersKey)
                );

        List<String> holders = new ArrayList<>();
        if (result != null) {
            for (Object obj : result) {
                if (obj instanceof String) {
                    holders.add((String) obj);
                }
            }
        }
        return holders;
    }

    /**
     * 动态调整最大并发数
     * 同时为holders_key设置相同的过期时间
     */
    public void updateMaxConcurrent(String limiterKey, int newMaxConcurrent) {
        Assert.hasText(limiterKey, "限流器键(limiterKey)不能为空");
        Assert.isTrue(newMaxConcurrent > 0, "最大并发数(newMaxConcurrent)必须大于0");

        String maxConcurrentKey = String.format(MAX_CONCURRENT_KEY_TEMPLATE, limiterKey);
        String holdersKey = String.format(HOLDERS_KEY_TEMPLATE, limiterKey);
        log.info("尝试更新最大并发数：key={}, newMaxConcurrent={}", maxConcurrentKey, newMaxConcurrent);

        List<Object> keys = Arrays.asList(maxConcurrentKey, holdersKey);
        Long defalutExpireDaysSeconds = TimeUnit.DAYS.toSeconds(MAX_CONCURRENT_KEY_DEFAULT_EXPIRE_DAYS);

        RScript rScript = redissonClient.getScript();
        rScript.eval(
                RScript.Mode.READ_WRITE,
                this.updateMaxConcurrentScript,
                RScript.ReturnType.INTEGER,
                keys,
                newMaxConcurrent,
                defalutExpireDaysSeconds.intValue()
        );
    }

    /**
     * 从业务数据恢复持有者记录
     * 同时为holders_key设置与max_key相同的过期时间
     */
    public void recoverHolders(String limiterKey, List<String> holderIds) {
        Assert.hasText(limiterKey, "限流器键(limiterKey)不能为空");
        Assert.notNull(holderIds, "持有者ID列表(holderIds)不能为空");

        String holdersKey = String.format(HOLDERS_KEY_TEMPLATE, limiterKey);
        // 1. 定义两个过期时间参数（均为 int 类型，避免 Lua 解析 nil）
        Long maxKeyExpireSeconds = TimeUnit.DAYS.toSeconds(MAX_CONCURRENT_KEY_DEFAULT_EXPIRE_DAYS); // holders_key 的过期时间
        Long holderExpireSeconds = TimeUnit.HOURS.toSeconds(HOLDER_DEFAULT_EXPIRE_HOURS);         // 每个 holder 的过期秒数

        if (holderIds.isEmpty()) {
            redissonClient.getKeys().delete(holdersKey);
            return;
        }

        // 2. 构建 Lua 脚本：参数顺序严格对应 ARGV[1]~ARGV[n]
        StringJoiner batchScript = new StringJoiner("\n");
        batchScript.add("local holders_key = KEYS[1]");
        batchScript.add("local max_key_expire = tonumber(ARGV[1])"); // ARGV[1]：holders_key 的过期秒数
        batchScript.add("local holder_expire_seconds = tonumber(ARGV[2])"); // ARGV[2]：holder 的过期秒数（修复 nil 问题）
        batchScript.add("local current_seconds = tonumber(redis.call('time')[1])");
        batchScript.add("local holder_expire_timestamp = current_seconds + holder_expire_seconds"); // 基于 ARGV[2] 计算过期时间
        batchScript.add("redis.call('del', holders_key)"); // 清空现有持有者

        // 3. 批量添加 holder（从 ARGV[3] 开始是 holderId）
        for (int i = 0; i < holderIds.size(); i++) {
            // ARGV[3 + i] 对应第 i 个 holderId
            batchScript.add(String.format("redis.call('zadd', holders_key, holder_expire_timestamp, ARGV[%d])", i + 3));
        }

        // 4. 为 holders_key 设置过期时间（使用 ARGV[1]）
        batchScript.add("redis.call('expire', holders_key, max_key_expire)");
        batchScript.add("return 1");

        List<Object> args = new ArrayList<>();
        args.add(maxKeyExpireSeconds.intValue());
        args.add(holderExpireSeconds.intValue());
        args.addAll(holderIds);

        // 执行脚本
        redissonClient.getScript().eval(
                RScript.Mode.READ_WRITE,
                batchScript.toString(),
                RScript.ReturnType.INTEGER,
                Collections.singletonList(holdersKey), // KEYS[1]
                args.toArray()
        );
    }

    /**
     * 定时清理任务 - 每5分钟执行一次
     * 清理所有过期的持有者记录
     */
    @Scheduled(fixedRate = CLEANUP_INTERVAL_MILLIS)
    public void cleanupExpiredHolders() {
        try {
            // 获取本地缓存的快照，避免在迭代时发生并发修改异常
            Map<String, ConcurrentHashMap<String, Long>> cacheSnapshot = activeHoldersCache.asMap();
            if (cacheSnapshot.isEmpty()) {
                log.debug("本地缓存中没有活跃的限流器Key，跳过本次清理。");
                return;
            }

            log.info("开始清理过期持有者记录，本地缓存中共有 {} 个活跃Key。", cacheSnapshot.size());
            int processedKeys = 0;
            int totalCleanedHolders = 0;

            RScript script = redissonClient.getScript();
            // 遍历本地缓存中的每一个 holdersKey
            for (Map.Entry<String, ConcurrentHashMap<String, Long>> entry : cacheSnapshot.entrySet()) {
                String holdersKey = entry.getKey();
                script.eval(
                        RScript.Mode.READ_WRITE,
                        cleanupScript,
                        RScript.ReturnType.INTEGER,
                        Collections.singletonList(holdersKey)
                );

                ConcurrentHashMap<String, Long> localHolderMap = entry.getValue();
                // 如果本地Map为空，跳过
                if (localHolderMap.isEmpty()) {
                    processedKeys++;
                    continue;
                }

                List<String> currentHolders = getCurrentHolders(holdersKey);
                try {
                    Set<String> validHolderIdsInRedis = new HashSet<>(currentHolders);

                    // 找出本地缓存中存在，但 Redis 中不存在的 Holder ID（这些是需要清理的）
                    Set<String> holdersToClean = new HashSet<>(localHolderMap.keySet());
                    holdersToClean.removeAll(validHolderIdsInRedis);

                    // 3. 执行清理：从本地缓存中删除这些 Holder
                    if (!holdersToClean.isEmpty()) {
                        log.info("Key '{}' 下发现 {} 个本地缓存Holder需要清理: {}", holdersKey, holdersToClean.size(), holdersToClean);
                        for (String holderId : holdersToClean) {
                            localHolderMap.remove(holderId);
                        }
                        totalCleanedHolders += holdersToClean.size();
                    }
                    processedKeys++;
                } catch (Exception e) {
                    // 确保单个 Key 的清理失败不会影响整个任务
                    log.error("清理 Key '{}' 时发生异常，已跳过。", holdersKey, e);
                }
            }

            log.info("本次清理完成，共处理了 {} 个Key，清理了 {} 个过期/无效的Holder。",
                    processedKeys, totalCleanedHolders);
        } catch (Exception e) {
            // 捕获任务级别的严重异常
            log.error("执行过期持有者清理任务时发生严重异常，任务已中断。", e);
        }
    }


    /**
     * 定时续期任务：为所有标记为需要续期的 Holder 执行批量续期
     */
    @Scheduled(fixedRate = RENEW_LEASE_INTERVAL_MILLIS)
    public void renewLeaseForActiveHolders() {
        // 1. 获取本地缓存的快照，避免在迭代过程中缓存结构发生变化导致并发问题
        Map<String, ConcurrentHashMap<String, Long>> cacheSnapshot = activeHoldersCache.asMap();

        if (cacheSnapshot.isEmpty()) {
            log.debug("本地缓存中没有需要续期的活跃 Holder，跳过本次续期任务。");
            return;
        }

        log.info("开始执行 Holder 续期任务，本地缓存中共有 {} 个活跃的 holders_key 需要检查。", cacheSnapshot.size());

        // 2. 遍历每个 holders_key
        for (Map.Entry<String, ConcurrentHashMap<String, Long>> entry : cacheSnapshot.entrySet()) {
            String holdersKey = entry.getKey();
            ConcurrentHashMap<String, Long> holderMap = entry.getValue();

            // 如果该 holders_key 下没有 Holder，直接跳过
            if (holderMap.isEmpty()) {
                continue;
            }

            // 3. 筛选出距离上次续期已超过间隔的 Holder
            List<String> holdersToRenew = new ArrayList<>();
            long currentTime = System.currentTimeMillis();
            long renewIntervalMillis = RENEW_INTERVAL_SECONDS * 1000; // 例如：30 * 60 * 1000 = 1800000ms

            // 使用迭代器进行安全遍历和移除
            Iterator<Map.Entry<String, Long>> iterator = holderMap.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, Long> holderEntry = iterator.next();
                String holderId = holderEntry.getKey();
                Long lastRenewTime = holderEntry.getValue();

                // 判断是否需要续期
                if (currentTime - lastRenewTime >= renewIntervalMillis) {
                    holdersToRenew.add(holderId);
                    // 先从本地缓存中移除，后续根据续期结果决定是否重新添加
                    iterator.remove();
                }
            }

            // 4. 如果有需要续期的 Holder，则执行批量续期
            if (!holdersToRenew.isEmpty()) {
                try {
                    // 从 holdersKey (e.g., "rate_limiter:holders:my_service") 中解析出 limiterKey (e.g., "my_service")
                    String limiterKey = holdersKey.replace(HOLDERS_KEY_TEMPLATE.replace("%s", ""), "");

                    log.debug("为 holders_key '{}' 下的 {} 个 Holder 执行批量续期: {}", holdersKey, holdersToRenew.size(), holdersToRenew);

                    // 调用批量续期方法
                    long successfullyRenewedCount = batchRenewLeases(limiterKey, holdersToRenew);

                    log.info("批量续期完成 - Key: '{}', 尝试续期: {}, 成功续期: {}", holdersKey, holdersToRenew.size(), successfullyRenewedCount);

                    // 5. 根据续期结果更新本地缓存
                    if (successfullyRenewedCount > 0) {
                        // 乐观更新：如果有任何一个续期成功，我们就认为该 Key 下的 Holder 仍然活跃
                        // 将所有尝试续期的 Holder 重新添加到本地缓存，并更新其续期时间戳
                        for (String holderId : holdersToRenew) {
                            holderMap.put(holderId, currentTime);
                        }
                        log.debug("已将 Key '{}' 下的 {} 个 Holder 重新添加到本地续期缓存。", holdersKey, holdersToRenew.size());
                    } else {
                        // 如果续期全部失败，说明这些 Holder 可能已在 Redis 中过期或被删除，无需再添加回本地缓存
                        log.warn("Key '{}' 下的 {} 个 Holder 续期全部失败，可能已过期或被删除。", holdersKey, holdersToRenew.size());
                    }

                } catch (Exception e) {
                    log.error("为 holders_key '{}' 执行批量续期时发生异常。待续期列表: {}", holdersKey, holdersToRenew, e);
                }
            }

            // 6. 清理：如果 holderMap 为空了，将其从顶层缓存中移除
            if (holderMap.isEmpty()) {
                try {
                    activeHoldersCache.invalidate(holdersKey);
                    log.debug("已将空的 holders_key '{}' 从活跃缓存中移除。", holdersKey);
                } catch (Exception e) {
                    log.error("尝试移除空的 holders_key '{}' 时发生异常。", holdersKey, e);
                }
            }
        }

        log.info("本次 Holder 续期任务执行完毕。");
    }

    /**
     * (这是之前定义的批量续期方法，为了完整性再次列出)
     * 调用批量续期 Lua 脚本为指定列表的 Holder 续期。
     * @param limiterKey 限流器键
     * @param holderIds 需要续期的 Holder ID 列表
     * @return 成功续期的数量
     */
    public long batchRenewLeases(String limiterKey, List<String> holderIds) {
        Assert.hasText(limiterKey, "限流器键(limiterKey)不能为空");
        Assert.notNull(holderIds, "holder_id 列表不能为空");
        if (holderIds.isEmpty()) {
            return 0;
        }

        String holdersKey = String.format(HOLDERS_KEY_TEMPLATE, limiterKey);
        Long expireSeconds = TimeUnit.HOURS.toSeconds(HOLDER_DEFAULT_EXPIRE_HOURS);

        List<Object> args = new ArrayList<>();
        args.add(expireSeconds.intValue());
        args.addAll(holderIds);

        RScript script = redissonClient.getScript();
        Long result = script.eval(
                RScript.Mode.READ_WRITE,
                batchRenewLeaseScript,
                RScript.ReturnType.INTEGER,
                Collections.singletonList(holdersKey),
                args.toArray()
        );

        return result != null ? result : 0;
    }

    /**
     * 关闭方法，清理资源
     */
    @PreDestroy
    public void shutdown() {
        log.info("分布式限流器已关闭");
    }
}